package com.ybw.rocketmq.demo.mq;

import com.ybw.rocketmq.demo.constant.MqConstant;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

/**
 * 设置消费模式consumeMode = ConsumeMode.ORDERLY，默认情况下是并发消费模式（ConsumeMode.CONCURRENTLY）
 *
 * @author ybwei
 * @date 2022/3/8 17:20
 **/
@Component("consumer")
@Slf4j
@RocketMQMessageListener(topic = MqConstant.TOPIC, consumerGroup = MqConstant.CONSUMER_GROUP, consumeMode = ConsumeMode.ORDERLY)
public class Consumer implements RocketMQListener<MessageExt> {


    @Override
    public void onMessage(MessageExt messageExt) {
        String msg = new String(messageExt.getBody());
        log.info("consumer start,msg:{}", msg);
        try {
            //模拟业务处理
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
            //抛异常，稍后消费
            throw new RuntimeException(e.getMessage());
        }
        log.info("consumer end,msg:{}", msg);
    }
}
